package cn.lagou.dw.flume.interceptor;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.interceptor.Interceptor;
import org.junit.Test;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.common.base.Charsets;

public class LogTypeInterceptor implements Interceptor {

	public void initialize() {
		// TODO Auto-generated method stub

	}

	public Event intercept(Event event) {
		// 获取 event 的 body
		String eventBody = new String(event.getBody(), Charsets.UTF_8);
		// 获取 event 的 header
		Map<String, String> headersMap = event.getHeaders();
		// 解析body获取json串
		String[] bodyArr = eventBody.split("\\s+");
		try {
			String jsonStr = bodyArr[6];
			// 解析json串获取时间戳
			String timestampStr = "";
			JSONObject jsonObject = JSON.parseObject(jsonStr);
			if (headersMap.getOrDefault("logtype", "").equals("start")) {
				// 取启动日志的时间戳
				JSONObject appJson = jsonObject.getJSONObject("app_active");
				timestampStr = appJson.getString("time");
			} else if (headersMap.getOrDefault("logtype", "").equals("event")) {
				// 取事件日志第一条记录的时间戳
				JSONArray jsonArray = jsonObject.getJSONArray("lagou_event");
				if (jsonArray.size() > 0) {
					timestampStr = jsonArray.getJSONObject(0).getString("time");
				}
			}
			// 将时间戳转换为字符串 "yyyy-MM-dd"
			// 将字符串转换为Long
			long timestamp = Long.parseLong(timestampStr);
			DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
			Instant instant = Instant.ofEpochMilli(timestamp);
			LocalDateTime localDateTime = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
			String date = formatter.format(localDateTime);
//将转换后的字符串放置header中
			headersMap.put("logtime", date);
			event.setHeaders(headersMap);
		} catch (Exception e) {
			e.printStackTrace();
			headersMap.put("logtime", "Unknown");
			event.setHeaders(headersMap);
		}
		return event;
	}

	public List<Event> intercept(List<Event> events) {
		List<Event> lstEvent = new ArrayList<Event>();
		for (Event event : events) {
			Event outEvent = intercept(event);
			if (outEvent != null) {
				lstEvent.add(outEvent);
			}
		}
		return lstEvent;
	}

	public void close() {
		// TODO Auto-generated method stub

	}
	public static class Builder implements Interceptor.Builder {
		public Interceptor build() {
			return new LogTypeInterceptor();
		}

		public void configure(Context context) {
		}

		}
//}


	@Test
	public void startJunit() {
		String str = "2020-08-02 18:19:32.959 [main] INFO com.lagou.ecommerce.AppStart - {\"app_active\":{\"name\":\"app_active\",\"json\":{\"entry\":\"1\",\"action\":\"0\",\"error_code\":\"0\"},\"time\":1596342840284},\"attr\":{\"area\":\"大庆\",\"uid\":\"2F10092A2\",\"app_v\":\"1.1.15\",\"event_type\":\"common\",\"device_id\":\"1FB872-9A1002\","
				+ "\"os_type\":\"2.8\",\"channel\":\"TB\",\"language\":\"chinese\",\"brand\":\"iphone-8\"}}";
//		String log = '2020-08-20 11:56:08.211 [main] INFO  com.lagou.ecommerce.AppStart - {"app_active":{"name":"app_active","json":{"entry":"2","action":"0","error_code":"0"},"time":1595288248066},"attr":{"area":"三门峡","uid":"2F10092A1","app_v":"1.1.0","event_type":"common","device_id":"1FB872-9A1001","os_type":"0.97","channel":"WM","language":"chinese","brand":"xiaomi-3"}}';
		Map<String, String> map = new HashMap<String, String>(); // new Event
		Event event = new SimpleEvent();
		map.put("logtype", "start");
		event.setHeaders(map);
		event.setBody(str.getBytes(Charsets.UTF_8));
// 调用interceptor处理event
		LogTypeInterceptor customerInterceptor = new LogTypeInterceptor();
		Event outEvent = customerInterceptor.intercept(event);
//处理结果
		Map<String, String> headersMap = outEvent.getHeaders();
		System.out.println(JSON.toJSONString(headersMap));
	}
}
